package com.zyc.serialize.kryo;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.zyc.remoting.dto.RpcRequest;
import com.zyc.remoting.dto.RpcResponse;
import com.zyc.serialize.Serializer;
import lombok.extern.slf4j.Slf4j;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;

/**
 * @author zyc
 * @create 2022-06-04-16:23
 */
public class KryoSerializer implements Serializer {

    /**
     * Kryo 线程不安全。所以用threadLocal存放
     */
    private static final ThreadLocal<Kryo> KRYO_THREAD_LOCAL = ThreadLocal.withInitial(() -> {
        Kryo kryo = new Kryo();
        kryo.register(RpcResponse.class);
        kryo.register(RpcRequest.class);
        kryo.setReferences(true);
        kryo.setRegistrationRequired(false);
        return kryo;
    });

    @Override
    public byte[] serialize(Object object) {
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        Output output = new Output(byteArrayOutputStream);
        Kryo kryo = KRYO_THREAD_LOCAL.get();
        kryo.writeObject(output, object);
        KRYO_THREAD_LOCAL.remove();
        return output.toBytes();
    }

    @Override
    public <T> T deserialize(byte[] bytes, Class<T> clazz) {
        ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes);
        Input input = new Input(byteArrayInputStream);
        Kryo kryo = KRYO_THREAD_LOCAL.get();
        Object object = kryo.readObject(input, clazz);
        KRYO_THREAD_LOCAL.remove();
        return clazz.cast(object);
    }
}
